package com.niit.MostPopularMovieDecadeStart1980;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
import java.util.Iterator;
import java.util.TreeMap;

public class MPMDSMapper extends Mapper<LongWritable, Text, Text, MPMDSBean>{
    private TreeMap<MPMDSBean, Text> flowMap80 = new TreeMap<>();
    private TreeMap<MPMDSBean, Text> flowMap90 = new TreeMap<>();
    private TreeMap<MPMDSBean, Text> flowMap00 = new TreeMap<>();
    private TreeMap<MPMDSBean, Text> flowMap10 = new TreeMap<>();

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        // 1. 获取一行
        String line = value.toString();

        // 2. 切割
        String[] movies = line.split("\t");

//        for (String s:movies){
//            System.out.println(s);
//        }

        // 3. 封装数据
        String rate = movies[0];
        String rank = movies[1];
        String types = movies[2];
        String regions = movies[3];
        String title = movies[4];
        String release_data = movies[5];
        String actor_count = movies[6];
        String vote_count = movies[7];
        String score = movies[8];
        String actors = movies[9];

        MPMDSBean flowBean = new MPMDSBean(rate,rank,types,regions,title,release_data,actor_count,vote_count,score,actors);
        Text movieInfo = new Text(rate+"\t"+rank+"\t"+types+"\t"+regions+"\t"+title+
                "\t"+release_data+"\t"+actor_count+"\t"+vote_count+"\t"+score+"\t"+actors);
        // 4. 向TreeMap中添加数据
        if(flowBean.getRelease_data().charAt(2) == '8') flowMap80.put(flowBean,movieInfo);
        if(flowBean.getRelease_data().charAt(2) == '9') flowMap90.put(flowBean,movieInfo);
        if(flowBean.getRelease_data().charAt(2) == '0' && flowBean.getRelease_data().charAt(0) == '2') flowMap00.put(flowBean,movieInfo);
        if(flowBean.getRelease_data().charAt(2) == '1' && flowBean.getRelease_data().charAt(0) == '2') flowMap10.put(flowBean,movieInfo);

        // 5. 限制TreeMap的数据量，超过10条就删除掉流量最小的一条数据
        if (flowMap80.size() > 10) {
            flowMap80.remove(flowMap80.lastKey());
        }
        if (flowMap90.size() > 10) {
            flowMap90.remove(flowMap90.lastKey());
        }
        if (flowMap00.size() > 10) {
            flowMap00.remove(flowMap00.lastKey());
        }
        if (flowMap10.size() > 10) {
            flowMap10.remove(flowMap10.lastKey());
        }

    }

    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        // 6 遍历treeMap集合，输出数据
        Iterator<MPMDSBean> iterator80 = flowMap80.keySet().iterator();
        Iterator<MPMDSBean> iterator90 = flowMap90.keySet().iterator();
        Iterator<MPMDSBean> iterator00 = flowMap00.keySet().iterator();
        Iterator<MPMDSBean> iterator10 = flowMap10.keySet().iterator();

        while (iterator80.hasNext()) {
            MPMDSBean valueOut = iterator80.next();
            Text keyOut = flowMap80.get(valueOut);
            context.write(keyOut, valueOut);
        }
        while (iterator90.hasNext()) {
            MPMDSBean valueOut = iterator90.next();
//            System.out.println(valueOut);
            Text keyOut = flowMap90.get(valueOut);
            context.write(keyOut, valueOut);
        }
        while (iterator00.hasNext()) {
            MPMDSBean valueOut = iterator00.next();
            Text keyOut = flowMap00.get(valueOut);
            context.write(keyOut, valueOut);
        }
        while (iterator10.hasNext()) {
            MPMDSBean valueOut = iterator10.next();
//            System.out.println(valueOut);
            Text keyOut = flowMap10.get(valueOut);
            context.write(keyOut, valueOut);
        }
    }
}
